package rtc.core.net.rpc;

import com.coco.base.log.SLog;

import java.net.InetSocketAddress;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

import rtc.core.net.CoreSocket;
import rtc.core.net.CoreSocketImpl;
import rtc.core.net.MsgBody;

/**
 * Created by Administrator on 2018/2/23.
 */

public class CoreNet {
    public static final long TIMEOUT_CONNECT = 15 * 1000;
    public static final long LOOP_SELECTOR_INTERVAL = 1000;
    private static final String TAG = "CoreNet";
    private static CoreNet INSTANCE = null;
    private CoreSocketImpl mSocket;
    private final ExecutorService mSendExecutor = new ThreadPoolExecutor(1, 1, 5 * 60 * 1000, TimeUnit.SECONDS,new LinkedBlockingQueue<Runnable>());
    private final ExecutorService mRecvExecutor = new ThreadPoolExecutor(1, 1, 5 * 60 * 1000, TimeUnit.SECONDS,new LinkedBlockingQueue<Runnable>());
    private DataCallback mDataCallback;
    public interface DataCallback{
        void onData(MsgBody msgBody);
    }

    public interface SendCallback{
        void onResult(int length);
    }

    private CoreNet(){
    }
    public static CoreNet getInstance(){
        if(INSTANCE == null){
            synchronized (CoreNet.class){
                if(INSTANCE == null){
                    INSTANCE = new CoreNet();
                }
            }
        }
        return INSTANCE;
    }

    void setDataCallback(DataCallback callback) {
        this.mDataCallback = callback;
    }

    private final CoreSocket.Callback mSocketCallback = new CoreSocket.Callback() {
        @Override
        public void onConnected() {
            SLog.i(TAG,"mSocketCallback onConnected");
        }

        @Override
        public void onDisconnected() {
            SLog.i(TAG,"mSocketCallback onDisconnected");
        }

        @Override
        public void onData(byte[] data) {
            SLog.i(TAG,"mSocketCallback onData data.length = "+data.length);
            mRecvExecutor.execute(new Runnable() {
                @Override
                public void run() {
                    try{
                        handleData(data);
                    }catch (Exception e){
                        e.printStackTrace();
                        SLog.e(TAG,"mSocketCallback onData handleData Exception",e);
                    }
                }
            });
        }

        @Override
        public void onSelectorLoop(int count) {
            //SLog.i(TAG,"mSocketCallback onSelectorLoop count = "+count);
        }
    };

    private void handleData(final byte[] bytes) throws Exception {
        if(mDataCallback != null){
            mDataCallback.onData(new MsgBody(bytes));
        }
    }

    public synchronized Boolean connect(boolean sync,String ip,int port){
        final long t0 = System.currentTimeMillis();
        if(mSocket != null){
            InetSocketAddress address = new InetSocketAddress(ip,port);
            if(mSocket.state() == CoreSocket.STATE_CONNECTED
                    && address.equals(mSocket.address())){
                SLog.w(TAG,"repeat connect same address,ip = %s,port = %s,sync = %s",ip,port,sync);
                return true;
            }else{
                mSocket.close();
                mSocket = null;
            }
        }
        mSocket = new CoreSocketImpl(ip,port,mSocketCallback);
        final Boolean[] result = new Boolean[]{false};
        final CountDownLatch latch = new CountDownLatch(1);
        final Runnable connRunnable;
        if(!sync){
            connRunnable = new Runnable() {
                @Override
                public void run() {
                    boolean connected = mSocket.connect(TIMEOUT_CONNECT);
                    SLog.i(TAG,"connect result = %s,expend %sms,sync = %s,ip = %s,port = %s,timeout = %s,",connected,(System.currentTimeMillis() - t0), false,ip,port,15000);
                    if(connected){
                        mSocket.loopSelector(LOOP_SELECTOR_INTERVAL);
                    }
                }
            };
        }else{
            connRunnable = new Runnable() {
                @Override
                public void run() {
                    boolean connected = mSocket.connect(TIMEOUT_CONNECT);
                    SLog.i(TAG,"connect result = %s,expend %sms,sync = %s,ip = %s,port = %s,timeout = %s,",connected,(System.currentTimeMillis() - t0), true,ip,port,15000);
                    result[0] = connected;
                    latch.countDown();
                    if(connected){
                        mSocket.loopSelector(LOOP_SELECTOR_INTERVAL);
                    }
                }
            };
        }
        new Thread(connRunnable,"SocketConn").start();
        if(sync){
            try {
                latch.await();
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            return result[0];
        }
        return null;
    }

    public synchronized void close(){
        if(mSocket != null && mSocket.state() != CoreSocket.STATE_DISCONNECTED){
            mSocket.close();
        }
    }

    public synchronized int send(final byte[] bytes,SendCallback callback){
        if(mSocket == null || mSocket.state() != CoreSocket.STATE_CONNECTED){
            SLog.e(TAG,"send() error,mSocket is null or mSocket not connected");
            if(callback != null){
                callback.onResult(0);
            }
            return 0;
        }
        if(callback == null){
            final Integer[] result = new Integer[]{0};
            final CountDownLatch latch = new CountDownLatch(1);
            mSendExecutor.execute(new Runnable() {
                @Override
                public void run() {
                    result[0] = mSocket.send(bytes,0,bytes.length);
                    latch.countDown();
                }
            });
            try {
                latch.await();
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            return result[0];
        }else{
            mSendExecutor.execute(new Runnable() {
                @Override
                public void run() {
                    int result = mSocket.send(bytes,0,bytes.length);
                    callback.onResult(result);
                }
            });
            return 0;
        }
    }

    public synchronized int state(){
        if(mSocket == null){
            return CoreSocket.STATE_DISCONNECTED;
        }
        return mSocket.state();
    }
}
